package com.example.agv.consumer;

import com.example.agv.constants.KafkaConstants;
import com.example.agv.models.AGVStatus;
import com.example.agv.models.Container;
import com.example.agv.models.WarehouseEntry;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;
import java.time.format.DateTimeFormatter;
import java.util.List;
import java.util.Random;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;

@Component
public class WsEntryListener {
    @Autowired
    private ExecutorService executer;

    @Autowired
    private KafkaTemplate<String, AGVStatus> kafkaTemplate;

	@Autowired
	private KafkaTemplate<String, Container> kafkaTemplate2;
    public String from_;

    @KafkaListener(
            topics = KafkaConstants.WS_ENTRY_TOPIC,
            groupId = KafkaConstants.GROUP_ID
    )
    public void listen(WarehouseEntry we_msg) {
        System.out.println("sending via kafka listener..");

        List<Container> containers = we_msg.getContainers();
        from_ = we_msg.getWarehouseEntryNumber();
        for (Container c : containers) {
        	executer.execute(new AGVTask(c));
        }
    }

    private class AGVTask implements Runnable {
    	Container entry_msg;

    	public AGVTask(Container container) {
    		entry_msg = container;
    	}

    	/*
    	 * {
			  "agv_id": "AGV-001",
			  "container_id": "CONTAINER123",
			  "status": "On",
			  "from":"WI-00-01",
			  "to":"WL-01-05",
			  "timestamp": "2023-03-15T12:34:56Z"
			}
    	 *
    	 */
        public void run() {
            //logger.debug("Thread # XXX is doing this task");//how to get thread id?
            int numThreads = 5;
            int randomSeconds = 5;
        	try {

				long threadId = Thread.currentThread().getId()%numThreads +1;
				String agv_id = "AGV-" + threadId;
				String container_id = entry_msg.getContainerNumber();
				String status = "ON";
				String from = from_;

		        // 获取当前的日期和时间
		        LocalDateTime currentDateTime = LocalDateTime.now();

		        // 定义一个日期时间格式化器
		        DateTimeFormatter formatter = DateTimeFormatter.ofPattern("yyyy-MM-dd HH:mm:ss");

		        // 使用格式化器将日期时间转换为字符串
		        String timestamp = currentDateTime.format(formatter);

				AGVStatus agv_msg = new AGVStatus(agv_id, container_id, status, from, "" , timestamp);

				//Sending the message to kafka topic queue，event=AGV载货
	            kafkaTemplate.send(KafkaConstants.AGV_TOPIC, agv_msg).get();

	            // 模拟AGV搬运过程，生成一个15到25之间的随机数
	            Random random = new Random();

	            // 生成一个0到6之间的随机数（包括0，不包括6）
	            randomSeconds = 15 + random.nextInt(25 - 15 + 1);
        		Thread.sleep(randomSeconds*1000);

        		from = "";
        		status = "OFF";
        		timestamp = LocalDateTime.now().format(formatter);
        		String to = entry_msg.getStorageLocation();
        		agv_msg = new AGVStatus(agv_id, container_id, status, "", to, timestamp);

        		//Sending the message to kafka topic queue， event=AGV卸货
	            kafkaTemplate.send(KafkaConstants.AGV_TOPIC, agv_msg).get();

				//传递的货物信息
				kafkaTemplate2.send(KafkaConstants.AGV_TOPIC, entry_msg).get();
				System.out.println("Container information: " + entry_msg);

			} catch (InterruptedException | ExecutionException e) {
				// TODO Auto-generated catch block
				e.printStackTrace();
			}
        }
    }

}
